package com.hudson.store.queue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author: 渔小鲤
 * @DateTime: 2025/7/12 0:09
 **/
public class TopicQueue {
    private static final Logger logger = LoggerFactory.getLogger(TopicQueue.class);

    private static final String TOPIC_QUEUE_PATH = "store/queue/data/";

    // 文件目錄
    private static File topicQueueDir;

    /**
     * 存放 queue 的FileChannel
     * String  Topic
     * Integer QueueId
     */
    private static final ConcurrentHashMap<String, ConcurrentHashMap<Integer, FileChannel>> queueMap = new ConcurrentHashMap<>();

    static {
        topicQueueDir = new File(TOPIC_QUEUE_PATH);
        if (!topicQueueDir.exists()) {
            topicQueueDir.mkdirs();
        }
        logger.debug("{}", topicQueueDir.exists());
        File[] files = topicQueueDir.listFiles();
        if (files != null) {
            for (File file : files) {
                final RandomAccessFile randomAccessFile;
                try {
                    String fileName = file.getName();

                    int lastDashIndex = fileName.lastIndexOf('-');
                    randomAccessFile = new RandomAccessFile(TOPIC_QUEUE_PATH + fileName, "rw");
                    String[] split = fileName.split("-");
                    ConcurrentHashMap<Integer, FileChannel> topicMap = queueMap.computeIfAbsent(fileName.substring(0, lastDashIndex), k -> new ConcurrentHashMap<>());

                    topicMap.computeIfAbsent(Integer.parseInt(fileName.substring(lastDashIndex + 1)), k -> randomAccessFile.getChannel());
                } catch (FileNotFoundException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    private static final ReentrantLock lock = new ReentrantLock();

    public static void addTopicQueue(List<String> topicList) {
        lock.lock();
        try {
            for (String topic : topicList) {
                ConcurrentHashMap<Integer, FileChannel> topicMap = queueMap.get(topic);
                if (Objects.isNull(topicMap)) {
                    if ("DLQ".equals(topic)) {
                        ConcurrentHashMap<Integer, FileChannel> map = new ConcurrentHashMap<>();
                        try {
                            map.put(0, new RandomAccessFile(TOPIC_QUEUE_PATH + topic, "rw").getChannel());
                        } catch (FileNotFoundException e) {
                            throw new RuntimeException(e);
                        }
                        queueMap.put(topic, map);
                        QueueNumUtil.putNum(topic, 1);
                        continue;
                    }
                    ConcurrentHashMap<Integer, FileChannel> map = new ConcurrentHashMap<>();
                    try {
                        map.put(0, new RandomAccessFile(TOPIC_QUEUE_PATH + topic + "-" + 0, "rw").getChannel());
                        map.put(1, new RandomAccessFile(TOPIC_QUEUE_PATH + topic + "-" + 1, "rw").getChannel());
                        map.put(2, new RandomAccessFile(TOPIC_QUEUE_PATH + topic + "-" + 2, "rw").getChannel());
                        map.put(3, new RandomAccessFile(TOPIC_QUEUE_PATH + topic + "-" + 3, "rw").getChannel());
                    } catch (FileNotFoundException e) {
                        logger.error("创建TopicQueue失败");
                        throw new RuntimeException(e);
                    }
                    queueMap.put(topic, map);
                    QueueNumUtil.putNum(topic);
                }
            }
        } finally {
            lock.unlock();
        }
    }

    public static void write(String topic, int queueId, long offset, int size, String tag) {
        FileChannel channel;
        ConcurrentHashMap<Integer, FileChannel> topicMap = queueMap.get(topic);
        if (Objects.isNull(topicMap)) {
            ConcurrentHashMap<Integer, FileChannel> map = new ConcurrentHashMap<>();
            if ("DLQ".equals(topic)) {
                try {
                    map.put(0, new RandomAccessFile(TOPIC_QUEUE_PATH + topic, "rw").getChannel());
                } catch (FileNotFoundException e) {
                    throw new RuntimeException(e);
                }
            } else {
                try {
                    map.put(0, new RandomAccessFile(TOPIC_QUEUE_PATH + topic + "-" + 0, "rw").getChannel());
                    map.put(1, new RandomAccessFile(TOPIC_QUEUE_PATH + topic + "-" + 1, "rw").getChannel());
                    map.put(2, new RandomAccessFile(TOPIC_QUEUE_PATH + topic + "-" + 2, "rw").getChannel());
                    map.put(3, new RandomAccessFile(TOPIC_QUEUE_PATH + topic + "-" + 3, "rw").getChannel());
                    QueueNumUtil.putNum(topic);
                } catch (FileNotFoundException e) {
                    logger.error("创建TopicQueue失败");
                    throw new RuntimeException(e);
                }
            }
            queueMap.put(topic, map);
            QueueNumUtil.putNum(topic, map.size());
            channel = map.get(queueId);
        } else {
            channel = topicMap.get(queueId);
        }

        // 计算 tag 的 hash 值（使用 Java 默认的 hashCode）
        long tagHash = tag.hashCode();

        // 构建 ByteBuffer
        ByteBuffer buffer = ByteBuffer.allocate(8 + 4 + 8); // 8(offset) + 4(size) + 8(tagHash)
        buffer.putLong(offset);
        buffer.putInt(size);
        buffer.putLong(tagHash);
        buffer.flip(); // 切换为读模式

        try {
            // ⬇️ 获取当前文件大小作为写入位置
            long position = channel.size();
            channel.write(buffer, position); // 从文件末尾开始写入
        } catch (IOException e) {
            throw new RuntimeException("Failed to write data to topic: " + topic, e);
        }
    }

    public static FileOffset read(String topic, int queueId, int num) {
        FileChannel channel = queueMap.get(topic).get(queueId);
        if (channel == null) {
            throw new RuntimeException("无法找到channel");
        }

        ByteBuffer buffer = ByteBuffer.allocate(8 + 4 + 8);
        try {
            channel.read(buffer, num * 20L);
            buffer.flip();

            if (buffer.remaining() < 20) {
                return null; // 数据不足，无法解析
            }

            long offset = buffer.getLong();
            int size = buffer.getInt();
            long tagHash = buffer.getLong();
            return new FileOffset(offset, size, tagHash);
        } catch (IOException e) {
            throw new RuntimeException("数据读取失败");
        }

    }

    public static class FileOffset {
        private long offset;
        private int size;

        private long tagHash;

        public FileOffset(long offset, int size, long tagHash) {
            this.offset = offset;
            this.size = size;
            this.tagHash = tagHash;
        }

        public long getOffset() {
            return offset;
        }

        public int getSize() {
            return size;
        }

        public long getTagHash() {
            return tagHash;
        }
    }
}
